package queue

import (
	"fmt"
	"runtime"
	"sync"
	"time"
)

/*
一个读 一个写
ReadWriteQueueTest time [559.5046ms] pps[1.787295] v[49999995000000]
ArrayQueueTest     time [357.0179ms] pps[2.800980] v[49999995000000]
MutexQueueTest     time [495.6454ms] pps[2.017571] v[49999995000000]
RingQueueTest      time [539.5571ms] pps[1.853372] v[49999995000000]
ChannelTest        time [733.0744ms] pps[1.364118] v[49999995000000]
*/

/*
多读 多写
RingQueueTestMulti goroutineCount[10] time [4.787169s] pps[2.088917] v[499999950000000]
ChannelTestMulti   goroutineCount[10] time[13.7637017s] pps[0.726549] v[499999950000000]

RingQueueTestMulti goroutineCount[100] time [46.3101671s] pps[2.159353] v[4999999500000000]
ChannelTestMulti   goroutineCount[100] time[2m27.3091016s] pps[0.678845] v[4999999500000000]
*/
var g_queue_cap_len uint32 = 131072
var g_queue_while_times int = 10000000

var g_queue_goroutine_count int = 10

func main() {
	ReadWriteQueueTest()
	ArrayQueueTest()

	MutexQueueTest()
	RingQueueTest()
	//ChannelTest()

	fmt.Println("============multi routine ===========")
	RingQueueTestMulti()
	ChannelTestMulti()
}

func ReadWriteQueueTest() {
	cqueue := NewReadWriteQueue(g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2)
	start := time.Now()
	gv := int64(0)

	// consum
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				v, ok := cqueue.Read()
				if !ok {
					runtime.Gosched()
					continue
				}
				gv += int64(v.(int))
				break
			}
		}
	}()

	// product
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				if ok := cqueue.Write(i); !ok {
					runtime.Gosched()
					continue
				}
				break
			}
		}
	}()

	wait.Wait()
	end := time.Since(start)

	fmt.Printf("ReadWriteQueueTest time [%s] pps[%f] v[%v]\n", end, (float64(g_queue_while_times)*100)/float64(end), gv)
}

func ArrayQueueTest() {
	cqueue := NewArrayQueue(g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2)
	start := time.Now()
	gv := int64(0)

	// consum
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				v, ok := cqueue.Read()
				if !ok {
					runtime.Gosched()
					continue
				}
				gv += int64(v.(int))
				break
			}
		}
	}()

	// product
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				if ok := cqueue.Write(i); !ok {
					runtime.Gosched()
					continue
				}
				break
			}
		}
	}()

	wait.Wait()
	end := time.Since(start)

	fmt.Printf("ArrayQueueTest     time [%s] pps[%f] v[%v]\n", end, (float64(g_queue_while_times)*100)/float64(end), gv)
}

func MutexQueueTest() {
	cqueue := NewMutexQueue(g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2)
	start := time.Now()
	gv := int64(0)

	// consum
	total := g_queue_while_times
	go func() {
		defer wait.Done()
		//for i := 0; i < g_queue_while_times; i++ {
		for {
			v, ok := cqueue.Read()
			if !ok {
				runtime.Gosched()
				continue
			}

			for _, x := range v {
				gv += int64(x.(int))
				total--
			}
			if total == 0 {
				break
			}
		}
		//}
	}()

	// product
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				if ok := cqueue.Write(i); !ok {
					runtime.Gosched()
					continue
				}
				break
			}
		}
	}()

	wait.Wait()
	end := time.Since(start)

	fmt.Printf("MutexQueueTest     time [%s] pps[%f] v[%v]\n", end, (float64(g_queue_while_times)*100)/float64(end), gv)
}

func RingQueueTest() {
	cqueue := NewRingQueue(g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2)
	start := time.Now()
	gv := int64(0)

	// consum
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				v, ok := cqueue.Read()
				if !ok {
					runtime.Gosched()
					continue
				}
				gv += int64(v.(int))
				break
			}
		}
	}()

	// product
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			for {
				if ok := cqueue.Write(i); !ok {
					runtime.Gosched()
					continue
				}
				break
			}
		}
	}()

	wait.Wait()
	end := time.Since(start)

	fmt.Printf("RingQueueTest      time [%s] pps[%f] v[%v]\n", end, (float64(g_queue_while_times)*100)/float64(end), gv)
}

func ChannelTest() {
	cqueue := make(chan interface{}, g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2)
	start := time.Now()
	gv := int64(0)

	// consum
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			v := <-cqueue
			gv += int64(v.(int))
		}
	}()

	// product
	go func() {
		defer wait.Done()
		for i := 0; i < g_queue_while_times; i++ {
			cqueue <- i
		}
	}()

	wait.Wait()
	end := time.Since(start)

	fmt.Printf("ChannelTest        time [%s] pps[%f] v[%v]\n", end, (float64(g_queue_while_times)*100)/float64(end), gv)
}

// ============multi routine ===========

func RingQueueTestMulti() {
	cqueue := NewRingQueue(g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2 * g_queue_goroutine_count)
	start := time.Now()
	gv := make([]int64, g_queue_goroutine_count)

	for index := 0; index < g_queue_goroutine_count; index++ {
		// consum
		go func(index int) {
			defer wait.Done()
			for i := 0; i < g_queue_while_times; i++ {
				for {
					v, ok := cqueue.Read()
					if !ok {
						runtime.Gosched()
						continue
					}
					gv[index] += int64(v.(int))
					break
				}
			}
		}(index)
	}

	for index := 0; index < g_queue_goroutine_count; index++ {
		// product
		go func() {
			defer wait.Done()
			for i := 0; i < g_queue_while_times; i++ {
				for {
					if ok := cqueue.Write(i); !ok {
						runtime.Gosched()
						continue
					}
					break
				}
			}
		}()
	}

	wait.Wait()
	end := time.Since(start)
	gvt := int64(0)
	for _, v := range gv {
		gvt += v
	}
	fmt.Printf("RingQueueTestMulti goroutineCount[%d] time [%s] pps[%f] v[%v]\n",
		g_queue_goroutine_count, end, (float64(g_queue_while_times*g_queue_goroutine_count)*100)/float64(end), gvt)
	//fmt.Println(gv)
}

func ChannelTestMulti() {
	cqueue := make(chan interface{}, g_queue_cap_len)
	var wait sync.WaitGroup
	wait.Add(2 * g_queue_goroutine_count)
	start := time.Now()
	gv := make([]int64, g_queue_goroutine_count)

	for index := 0; index < g_queue_goroutine_count; index++ {
		// consum
		go func(index int) {
			defer wait.Done()
			for i := 0; i < g_queue_while_times; i++ {
				v := <-cqueue
				gv[index] += int64(v.(int))
			}
		}(index)
	}
	for index := 0; index < g_queue_goroutine_count; index++ {
		// product
		go func() {
			defer wait.Done()
			for i := 0; i < g_queue_while_times; i++ {
				cqueue <- i
			}
		}()
	}
	wait.Wait()
	end := time.Since(start)
	gvt := int64(0)
	for _, v := range gv {
		gvt += v
	}
	fmt.Printf("ChannelTestMulti goroutineCount[%d] time[%s] pps[%f] v[%v]\n",
		g_queue_goroutine_count, end, (float64(g_queue_while_times*g_queue_goroutine_count)*100)/float64(end), gvt)
}
